package com.shujia.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo03WindowOpt {
  def main(args: Array[String]): Unit = {
    /**
     * 微博热搜：每隔1个小时，基于最近24小时内的搜索量等其它指标进行热度的计算
     * 简化：每隔5s钟，统计最近10s内的数据
     * 需要通过窗口实现：
     * 1、滚动窗口：统计周期时间和决定数据批次大小的时间相等
     * 2、滑动窗口：统计周期时间 小于 决定数据批次大小的时间
     */
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo03WindowOpt")
    conf.setMaster("local[2]")

    val ssc: StreamingContext = new StreamingContext(new SparkContext(conf), Durations.seconds(5))
    ssc.checkpoint("spark/data/ck")

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKeyAndWindow((i1: Int, i2: Int) => {
        i1 + i2
      }, (i1: Int, i2: Int) => {
        i1 - i2
      }, Durations.seconds(10), Durations.seconds(5))
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}
